package com.patsnap.data.npd.dw.etl.utils;

import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.model.*;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * sqs client for aws<br>
 * default with region: US_EAST_1
 * @author lwying
 * 
 */
public class AWSSQSUtils {
    
    private static Logger logger = Logger.getLogger(AWSSQSUtils.class);
    
    private static final int BATCH_SIZE = 10;
    
    /**
     * In the instance profile credentials, which exist within the instance
     * metadata associated with the IAM role for the EC2 instance.
     */
    private AWSSQSUtils() {
    }
    
    /**
     * get queue url as "https://sqs.us-east-1.amazonaws.com/[aws accountid]/[queue_name]"
     * @param sqs client
     * @param queue_name queue name
     * @return queue url
     */
    public static String getQueueURL(AmazonSQS sqs, String queue_name) {
        GetQueueUrlRequest request = new GetQueueUrlRequest(queue_name);
        String queue_url = sqs.getQueueUrl(request).getQueueUrl();
        return queue_url;
    }
    
    /**
     * batch get message
     * @param sqs client
     * @param queue_name queue name
     * @param batchSize batch size, limit max to 10
     * @return messages
     */
    public static List<Message> getMessages(AmazonSQS sqs, String queue_name, int batchSize) {
        String url = getQueueURL(sqs, queue_name);
        return getMessagesByQueueUrl(sqs, url, batchSize);
    }
    
    /**
     * batch get message
     * @param sqs client
     * @param queueUrl queue url
     * @param batchSize batch size
     * @return messages
     */
    public static List<Message> getMessagesByQueueUrl(AmazonSQS sqs, String queueUrl, int batchSize) {
        ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(queueUrl);
        receiveMessageRequest.setMaxNumberOfMessages(batchSize);
        ReceiveMessageResult result = sqs.receiveMessage(receiveMessageRequest);
        if (logger.isDebugEnabled()) {
            logger.debug(queueUrl + " get msg:" + result.getMessages().size());
        }
        return result.getMessages();
    }
    
    /**
     * get single message without delete, <b>should delete manually if need</b>
     * @param sqs client
     * @param queue_name queue name
     * @return message
     */
    public static Message getMessage(AmazonSQS sqs, String queue_name) {
        return getMessage(sqs, queue_name, false);
    }
    
    /**
     * get single message, <b>with deleted flag when return</b>
     * @param sqs client
     * @param queue_name queue name 
     * @param isDel auto delete or not, delete msg or not, true is delete
     * @return message
     */
    public static Message getMessage(AmazonSQS sqs, String queue_name, boolean isDel) {
        Message msg = null;
        List<Message> msgs = getMessages(sqs, queue_name, 1);
        if (CollectionUtils.isNotEmpty(msgs)) {
            msg = msgs.get(0);
            if (isDel) {
                delMsg(sqs, queue_name, msg);
            }
        }
        return msg;
    }
    
    /**
     * send message
     * @param sqs client
     * @param queue_name queue name
     * @param content message body 
     * @return status
     */
    public static boolean sendMsg(AmazonSQS sqs, String queue_name, String content) {
        return sendMsg(sqs, queue_name, content, null);
    }


    
    /**
     * send message to fifo queue
     * @param sqs client
     * @param queue_name queue name
     * @param content message body
     * @param messageGroupId group id, must not be null
     * @return status
     */
    public static boolean sendFIFOMsg(AmazonSQS sqs, String queue_name, String content, String messageGroupId) {
        return sendMsg(sqs, queue_name, content, messageGroupId);
    }
    
    /**
     * send message to fifo queue
     * @param sqs client
     * @param queue_name queue name
     * @param content message body
     * @param messageDeduplicationId messageDeduplicationId
     * @param messageGroupId group id, must not be null
     * @return status
     */
    public static boolean sendFIFOMsgWithDeduplicationId(AmazonSQS sqs, String queue_name, String content,
        String messageDeduplicationId, String messageGroupId) {
        SendMessageRequest request = new SendMessageRequest(getQueueURL(sqs, queue_name), content);
        if (messageDeduplicationId != null) {
            request.setMessageDeduplicationId(messageDeduplicationId);
        }
        if (messageGroupId != null) {
            request.setMessageGroupId(messageGroupId);
        }
        SendMessageResult result = sqs.sendMessage(request);
        if (logger.isDebugEnabled()) {
            logger.debug(queue_name + " send msg:" + content);
        }
        return result != null && result.getMessageId() != null;
    }
    
    /**
     * send message, if messageGroupId is not null, treat queue as FIFO queue.
     * @param sqs client
     * @param queue_name queue name
     * @param content message body
     * @param messageGroupId group id, need if fifo queue
     * @return status
     */
    public static boolean sendMsg(AmazonSQS sqs, String queue_name, String content, String messageGroupId) {
        SendMessageRequest request = new SendMessageRequest(getQueueURL(sqs, queue_name), content);
        if (messageGroupId != null) {
            request.setMessageGroupId(messageGroupId);
        }
        int retry = 3;
        while(retry-- > 0){
            SendMessageResult result = sqs.sendMessage(request);
            if (logger.isDebugEnabled()) {
                logger.debug(queue_name + " send msg:" + content);
            }
            boolean isSuccess = result != null && result.getMessageId() != null;
            if(isSuccess){
                return true;
            }else{
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
            }
        }
        logger.error("msg retry 3 fail to " + queue_name);
        return false;
        
    }
    
    /**
     * batch send message
     * @param sqs client
     * @param queue_name queue name
     * @param contents 
     * @return status
     */
    public static boolean batchSendMsg(AmazonSQS sqs, String queue_name, String... contents) {
        return batchSendMsg(sqs, queue_name, null, contents);
    }
    
    /**
     * batch send message to FIFO
     * @param sqs client
     * @param queue_name queue name
     * @param messageGroupId fifo group id 
     * @param contents texts
     * @return
     */
    public static boolean batchSendFIFOMsg(AmazonSQS sqs, String queue_name, String messageGroupId,
        String... contents) {
        return batchSendMsg(sqs, queue_name, messageGroupId, contents);
    }
    
    /**
     * batch send, default batch size 10
     * @param sqs client
     * @param queue_name queue name
     * @param msgs list
     * @return status
     */
    public static boolean batchSendMsg(AmazonSQS sqs, String queue_name, List<String> msgs) {
        return batchSendMsg(sqs, queue_name, msgs, null);
    }
    
    /**
     * batch send to FIFO queue
     * @param sqs client
     * @param queue_name queue name
     * @param messageGroupId groupid
     * @param msgs list
     * @return status
     */
    public static boolean batchSendFIFOMsg(AmazonSQS sqs, String queue_name, String messageGroupId, List<String> msgs) {
        return batchSendMsg(sqs, queue_name, msgs, messageGroupId);
    }
    
    /**
     * batch send message
     * @param sqs client
     * @param queue_name queue name
     * @param msgs message list
     * @param messageGroupId only for <a href="https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/FIFO-queues.html">FIFO queue</a>
     * @return status
     */
    public static boolean batchSendMsg(AmazonSQS sqs, String queue_name, List<String> msgs, String messageGroupId) {
        boolean isSuccess = true;
        List<String> batch = new ArrayList<>();
        for (String line : msgs) {
            batch.add(line);
            if (batch.size() >= BATCH_SIZE) {
                isSuccess &= batchSendMsg(sqs, queue_name, messageGroupId, batch.toArray(new String[] {}));
                batch.clear();
            }
        }
        //last
        if (batch.size() > 0) {
            isSuccess &= batchSendMsg(sqs, queue_name, messageGroupId, batch.toArray(new String[] {}));
            batch.clear();
        }
        return isSuccess;
    }
    
    /**
     * batch send message
     * @param sqs client
     * @param queue_name queue name
     * @param messageGroupId only for <a href="https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/FIFO-queues.html">FIFO queue</a>
     * @param contents array
     * @return status
     */
    public static boolean batchSendMsg(AmazonSQS sqs, String queue_name, String messageGroupId, String... contents) {
        SendMessageBatchRequest request = new SendMessageBatchRequest(getQueueURL(sqs, queue_name));
        List<SendMessageBatchRequestEntry> list = new ArrayList<SendMessageBatchRequestEntry>();
        int index = 0;
        for (String line : contents) {
            SendMessageBatchRequestEntry en = new SendMessageBatchRequestEntry();
            en.setMessageBody(line);
            if (messageGroupId != null) {
                en.setMessageGroupId(messageGroupId);
            }
            en.setId(String.valueOf(index++));
            list.add(en);
        }
        // set
        request.setEntries(list);
        SendMessageBatchResult result = sqs.sendMessageBatch(request);
        if (logger.isDebugEnabled()) {
            logger.debug(queue_name + " send msg:" + contents.length);
        }
        boolean isSuccess = result != null && CollectionUtils.isEmpty(result.getFailed());
        return isSuccess;
    }
    
    /**
     * delete message
     * @param sqs client
     * @param queue_name queue name
     * @param msg messge object
     */
    public static void delMsg(AmazonSQS sqs, String queue_name, Message msg) {
        String messageRecieptHandle = msg.getReceiptHandle();
        delMsg(sqs, queue_name, messageRecieptHandle);
    }
    
    /**
     * batch delete message
     * @param client
     * @param queue
     * @param msgs
     * @return
     */
    public static int batchDelete(AmazonSQS client, String queue, List<Message> msgs) {
        if (CollectionUtils.isEmpty(msgs)) {
            return 0;
        }
        DeleteMessageBatchRequest request = new DeleteMessageBatchRequest(queue);
        List<DeleteMessageBatchRequestEntry> ens = new ArrayList<>();
        for (Message msg : msgs) {
            DeleteMessageBatchRequestEntry en = new DeleteMessageBatchRequestEntry();
            en.setId(msg.getMessageId());
            en.setReceiptHandle(msg.getReceiptHandle());
            ens.add(en);
        }
        request.setEntries(ens);
        DeleteMessageBatchResult rs = client.deleteMessageBatch(request);
        return rs.getSuccessful().size();
    }
    
    /**
     * delete message with <b>queueURL</b>, not queue name!!<br>
     * @param sqs sqs  client
     * @param queueUrl {@link AWSSQSUtils#getQueueURL}
     * @param msg message object
     */
    public static void delMsgByByQueueUrl(AmazonSQS sqs, String queueUrl, Message msg) {
        String messageRecieptHandle = msg.getReceiptHandle();
        DeleteMessageRequest request = new DeleteMessageRequest(queueUrl, messageRecieptHandle);
        sqs.deleteMessage(request);
        if (logger.isDebugEnabled()) {
            logger.debug(queueUrl + " del msg:" + msg.getBody());
        }
    }
    
    /**
     * delete message
     * @param sqs client
     * @param queue_name queue name
     * @param msg message handle value {@link Message#receiptHandle}
     * @return
     */
    public static void delMsg(AmazonSQS sqs, String queue_name, String messageRecieptHandle) {
        DeleteMessageRequest request = new DeleteMessageRequest(getQueueURL(sqs, queue_name), messageRecieptHandle);
        sqs.deleteMessage(request);
        if (logger.isDebugEnabled()) {
            logger.debug(queue_name + " del msg:" + messageRecieptHandle);
        }
    }
    
    /**
     * get unread size
     * @param sqs client
     * @param queue_name queue name
     * @return queue size not processed
     */
    public static long getQueueUnreadSize(AmazonSQS sqs, String queue_name) {
        return getQueueSizeWithKey(sqs, queue_name, QueueAttributeName.ApproximateNumberOfMessages);
    }
    
    //    @Deprecated
    //    public long getQueueSizeWithKey(String queue_name, QueueAttributeName name) {
    //        return 0L;
    //    }
    
    /**
     * get size by key
     * @param sqs client
     * @param queue_name queue name
     * @param name attribute name, {@link QueueAttributeName}
     * @return long value
     */
    public static long getQueueSizeWithKey(AmazonSQS sqs, String queue_name, QueueAttributeName name) {
        GetQueueAttributesRequest request = new GetQueueAttributesRequest(getQueueURL(sqs, queue_name));
        String key = name.name();
        request.withAttributeNames(key);
        GetQueueAttributesResult result = sqs.getQueueAttributes(request);
        Map<String, String> rs = result.getAttributes();
        long size = 0;
        if (rs.containsKey(key)) {
            String s = rs.get(key);
            if (StringUtils.isNumeric(s)) {
                size = Long.parseLong(s);
            }
        }
        return size;
    }
}
